LangGraphで AIエージェントをまなんでいく - その1 はじめてのグラフ-

AIエージェントはユーザーとのやり取りを通じて特定の目標を達成することに焦点を当てています。
人間のように自律的に仕事をやってくれるAIのプログラムの総称のようなものですかね。

生成AIとの違いは、生成AIは主にコンテンツ生成に特化し、与えられた入力に対して新たなデータやコンテンツを作成することに特化していますね。生成されたコンテンツの修正をする場合、人間が都度やりとりして最終的なコンテンツを作っていくことになります。

AIエージェントは生成AIなどのAI技術を組み合わせて自律的に作業を実行してもらうシステムと言えるのではないでしょうか。
プログラムの作成からテスト、バグ修正を自動的に行う、顧客のリストから過去のやり取りを参照しつつメールを作成して送信する　といったことが1回の入力で済んでしまうといったことが可能になりつつあります。

LangGraphについて

LangGraph

LangGraphは、LLM（大規模言語モデル）を使ってステートフルなマルチアクターアプリケーションを構築するためのライブラリで、エージェントやマルチエージェントワークフローを作成するために使われます。

※ MITライセンス

LangGraph の特徴

• サイクルと分岐: アプリケーション内でループや条件分岐を実装できます。
• 永続性: グラフ内の各ステップ後に状態を自動的に保存します。グラフの実行を任意の時点で一時停止・再開でき、エラー復旧、人間を介したワークフロー、タイムトラベルなどをサポートします。
• 人間を介したプロセス: グラフの実行を中断し、エージェントが計画した次のアクションを承認または編集することができます。
• ストリーミング対応: 各ノードで生成される出力を、（トークンストリーミングを含めて）リアルタイムでストリーミングできます。
• LangChainとの統合: LangGraphはLangChainやLangSmithとシームレスに統合できます(必須ではない)

LangGraphでのアプリケーションの制御フロー

LangGraphは、LLM（大規模言語モデル）を活用したアプリケーションの制御フローを、Nodes（ノード）とEdges（エッジ）で表現することで高い制御性を提供します。

エージェントのワークフローを__グラフ__としてモデル化していて、以下の3つの主要コンポーネントでエージェントの動作を定義します。

• State（状態）: アプリケーションの現在のスナップショットを表す共有データ構造です。任意のPython型を使用できますが、通常はTypedDictやPydanticのBaseModelが用いられます。
• Nodes（ノード）: エージェントのロジックをエンコードするPython関数です。現在のStateを入力として受け取り、計算や副作用を実行し、更新されたStateを返します
• Edges（エッジ）: 現在のStateに基づいて、次に実行するNodeを決定するPython関数です。条件分岐や固定的な遷移を含むことができます。

これらのNodesとEdgesを組み合わせることで、時間とともにStateが進化する複雑なループワークフローを作成できます。
特に、LangGraphはこのStateの管理に強みを持っています。
NodesとEdgesはPython関数に過ぎず、その中には大規模言語モデル（LLM）や通常のPythonコードを含めることができます。

簡単なグラフを作ってみる

どのような動作をするのか試すため、
LangChain AcademyのLangGraph入門コース　にあるレッスンをやってみます。

これは3つのノードと1つの条件付きエッジを持つ単純なグラフです。

ライブラリのインストール

pip install --quiet -U langgraph

Stateの定義

from typing_extensions import TypedDict

class State(TypedDict):
  graph_state: str

グラフ内のすべてのノードとエッジの入力スキーマになるStateの定義をしています

Nodesの実装

def node_1(state):
    print("---Node 1---")
    return {"graph_state": state['graph_state'] +" I am"}

def node_2(state):
    print("---Node 2---")
    return {"graph_state": state['graph_state'] +" happy!"}

def node_3(state):
    print("---Node 3---")
    return {"graph_state": state['graph_state'] +" sad!"}

stateを引数にもつ関数の作成です。
上記の図のグラフを作成したいので、3つ作成しています。

Edgesの実装

import random
from typing import Literal

def decide_mood(state) -> Literal["node_2", "node_3"]:

    # Often, we will use state to decide on the next node to visit
    user_input = state['graph_state'] 

    # Here, let's just do a 50 / 50 split between nodes 2, 3
    if random.random() < 0.5:

        # 50% of the time, we return Node 2
        return "node_2"

    # 50% of the time, we return Node 3
    return "node_3"

Nodesを繋ぐEdgeの作成です。
ここでは、ノード_1からノード_2、ノード_3へ移動するためのものです。
条件によって移動するノードを決めています（条件付きエッジにする）

• 条件付きエッジ（Conditional Edges）は、ノード間を任意にルーティングしたい場合に使用します。
• 条件付きエッジは、何らかのロジックに基づいて次に訪れるノードを返す関数として実装されます

グラフの作成

from IPython.display import Image, display
from langgraph.graph import StateGraph, START, END

# Build graph
builder = StateGraph(State)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)

# Logic
builder.add_edge(START, "node_1")
builder.add_conditional_edges("node_1", decide_mood)
builder.add_edge("node_2", END)
builder.add_edge("node_3", END)

# Add
graph = builder.compile()

# View
display(Image(graph.get_graph().draw_mermaid_png()))

上で定義したコンポーネントからグラフを構築しています。

実行した結果を表示させてみました。
node_1から点線で他のノードに別れていますが、点線の場合は条件分岐のようですね。

グラフを呼び出す

graph.invoke({"graph_state" : "Hi, this is Lance."})

• invokeが呼ばれると、グラフはSTARTノードから実行を開始
• 定義されたノード(node_1, node_2, node_3)を順に進む。
• 条件付きエッジは、50/50の決定ルールを使用して、ノード1からノード2またはノード3へと移動。
• 各ノード関数は現在の状態を受け取り、グラフの状態を上書きして、新しい値を返す。
• 実行はENDノードに達するまで実行されます。

invokeはグラフを同期的に実行します。

実行結果

Stateが上書きして返されていますね。

参考

ライブラリのリファレンス